package fundamental;


import java.io.IOException;
import java.util.*;

import core.MoveOrNot;
import log.WriteToLog;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.util.Bytes;
import singleobject.MemoryInfo;


/**
 * Created by yi on 2017/7/4.
 */
public class Region {
    //three private member
    private  Map<String,String> regionRS=new HashMap<String,String>();
    private  Map<String,List<String>> regionTask=new HashMap<String,List<String>>();
    private  Map<String,String> regionKey=new HashMap<String,String>();
    //regionTable行键为table,value为regionName
    private  Map<String,List<String>> regionTable=new HashMap<String,List<String>>();
    private  volatile Map<String,Queue<Integer>> requestTimesQueue = new HashMap<String, Queue<Integer>>();

//填充三个map
    public void getRegionToRegionServer(String RS) throws IOException, InterruptedException {
        Admin admin = hbase.HBaseUtil.getHBASECONFIG().getConnection().getAdmin();
        List<HRegionInfo> hRegionInfos = admin.getOnlineRegions(ServerName.valueOf(RS));
        for(HRegionInfo hRegionInfo : hRegionInfos){
         MemoryInfo.getREGION().addRegionRS(Bytes.toString(hRegionInfo.getRegionName()),RS);
         MemoryInfo.getREGION().addRegionKey(Bytes.toString(hRegionInfo.getRegionName()),Bytes.toString(hRegionInfo.getStartKey())+","+Bytes.toString(hRegionInfo.getEndKey()));
         MemoryInfo.getREGION().addregionTable( hRegionInfo.getTable().toString(),Bytes.toString(hRegionInfo.getRegionName()));
//            hRegionInfo.
        }
//        System.out.println("regionRS:");
//        for (Map.Entry<String, String> entry : regionRS.entrySet()) {
//            System.out.println("key= " + entry.getKey() + " and value= " + entry.getValue());
//        }
        System.out.println("regionKey:");
        for (Map.Entry<String, String> entry : regionKey.entrySet()) {
            System.out.println("key= " + Arrays.toString(entry.getKey().getBytes()) + " and value= " + Arrays.toString(entry.getKey().getBytes()));
        }
//        System.out.println("regionTable:");
//        for (Map.Entry<String, List<String>> entry : regionTable.entrySet()) {
//            System.out.println("key= " + entry.getKey() + " and value= " + entry.getValue());
//        }



        admin.close();
}


    public void  init() throws IOException, InterruptedException{
        WriteToLog.write("init start...........\n");
        Admin admin = hbase.HBaseUtil.getHBASECONFIG().getConnection().getAdmin();
        ClusterStatus status =admin.getClusterStatus();
        for(ServerName server:status.getServers()){
            MemoryInfo.getRegionServer().rsNameInCluster.add(server.toString());
            WriteToLog.write("init!!!!!!!Write the Information of "+server+":\n");
           // System.out.println("serverName: "+server);
            getRegionToRegionServer(server.getServerName());
        }
        admin.close();
    }

    /*regionRS相关函数*/

    //韩伟娜使用，在开机初始状态填充此MAP
    //张欣使用，在移动region时使用
    public void addRegionRS(String rname,String RSname){
            regionRS.put(rname,RSname);
            WriteToLog.write("region:"+rname+" is add to RegionServer:"+RSname+" \n");


    }
    //张欣使用
    public String getTheRegionRS(String rname) {
        return regionRS.get(rname) ;
    }

    /*regionTask相关函数*/

    //添加一个region的task队列成员
    // 陈祥使用，在task到来时填充regionTask这个map
    //具体实现为：在任务到来时，拿到taskID以及该task要用的tableName,和task要用到的startKey,EndKey.然后通过regionKey和regionTable这两个map解析出task要用的region的名字。
    public String searchregionTask(String tableName,String startKey,String endKey){
       List<String> list= regionTable.get(tableName);
       //rname为table中所有的region名字
        for (String rname : list) {
            //对每一个region求出它的startkey和endkey.
            String startEnds = regionKey.get(rname);
            System.out.println("startEnds:"+startEnds);
            String[] startEnd = startEnds.split(",");
//            System.out.println("startEnd.length: "+startEnd.length);
//            System.out.println("startEnd[0].length: "+startEnd[0].length());
//            System.out.println("startEnd[1].length: "+startEnd[1].length());
//            System.out.println("startEnd.length: "+startEnd.length);
            byte[] taskS = startKey.getBytes();
            byte[] taskE = endKey.getBytes();
//            System.out.println("regionS:"+startEnd[0]);
//            System.out.println("regionE:"+startEnd[1]);
            System.out.println("taskS:"+startKey);
            System.out.println("taskE:"+endKey);

            if (startEnd.length==2) {
                if (startEnd[0].length() != 0) {
                    byte[] regionS = startEnd[0].getBytes();
                    byte[] regionE = startEnd[1].getBytes();
                    //taskS和taskE为task要用的startKey和endkey
                    int s = Bytes.compareTo(regionS, taskS);
                    int e = Bytes.compareTo(regionE, taskE);
                    if ((s<=0) && (e >= 0)) {
                        return rname;
                    }
                } else {
                    WriteToLog.write("start key is none");
                    byte[] regionE = startEnd[1].getBytes();
                    int e = Bytes.compareTo(regionE, taskE);
                    if (e >=0) {
                        WriteToLog.write("get the rname!");
                        return rname;
                    }
                }
            } else {
                byte[] regionS = startEnd[0].getBytes();
                int s = Bytes.compareTo(regionS, taskS);
                if((s<=0)){
                    return rname;
                }
            }
        }

        return "false";

    }
    public Map<String,List<String>> getRegionTask(){
        return regionTask;
    }
    public void addRegionTask(String taskID,String tableName,String startKey,String endKey){
        WriteToLog.write("Task:"+taskID+" come,tableName: "+tableName+" StartKey: "+startKey+" EndKey: "+endKey+"\n");
        String rname=searchregionTask(tableName, startKey, endKey);
        if(rname.equals("false")){
            WriteToLog.write("the task ask region is not exist");
            return;
        }
        List<String> list=new ArrayList<String>();
        if(regionTask.get(rname)!=null){
            list=regionTask.get(rname);
            list.add(taskID);
        }else{
            list.add(taskID);
        }

        regionTask.put(rname,list);
        System.out.println("when task comes,tasklist: "+list);
        WriteToLog.write("when task comes,tasklist: "+list);
        Boolean b =MemoryInfo.getTASK().isDataLocaility(taskID,tableName,startKey,endKey);
        if(b==true){
            WriteToLog.write("decide not to call the core code.");
        }else{
            WriteToLog.write("decide to call the core code.");
            new MoveOrNot().core(taskID,rname);
        }
    }
    //删除一个region的task队列成员
    //不知道谁使用，在一个task结束时使用。
    public void removeRegionTask(String taskID){
        String rname=new String();
        for (Map.Entry<String, List<String>> entry : regionTask.entrySet()) {
            if(entry.getValue().contains(taskID)){
                rname=entry.getKey();
            }
        }
        if(rname==null){
            System.out.println("the task is not exist,so cann't remove it from map regionTask");
            return;
        }
        System.out.println("when remove:taskID: "+taskID);
        WriteToLog.write("when remove:taskID: "+taskID);
        List<String> list=regionTask.get(rname);
        System.out.println("when remove:tasklist: "+list.toString());
        WriteToLog.write("when remove:tasklist: "+list.toString());
        list.remove(taskID);
        regionTask.put(rname,list);
        WriteToLog.write("Task:"+taskID+"finish,region:"+rname+" is not been used\n");
    }
    //张欣使用
    public List<String> getTheRegionTask(String rname) {
        return regionTask.get(rname) ;
    }

    /*regionKey相关函数*/

    //初始化region的startKey 和EndKey
    //韩伟娜使用
    public void addRegionKey(String rname,String key){
        regionKey.put(rname,key);


    }
    //韩伟娜使用，在任务到来时计算task所用到的region时使用。
    public String getTheRegionKey(String rname) {
        return regionKey.get(rname) ;
    }

    /*regionTable相关函数*/
    //初始状态使用，一个表中对应的region
    public void addregionTable(String tableName,String regionName){
        List<String> list;
       if(regionTable.get(tableName)==null){
         list=new ArrayList<String>();
       }else{
         list=regionTable.get(tableName);
       }
         list.add(regionName);
        regionTable.put(tableName,list);
    }
    /*requestTimesQueue相关函数*/
    //每秒更新，一个RS的Queue队列
    //陈祥使用
    private void flashOneQueue(String RS,int t){
        Queue<Integer> queue=new LinkedList<Integer>();
        if(requestTimesQueue.get(RS)==null){
              // queue =new LinkedList<Integer>();
            requestTimesQueue.put(RS,queue);
            queue.offer(0);
            queue.offer(0);
            queue.offer(0);
            queue.offer(0);
            queue.offer(0);
        }
//                queue.clear();
        queue=requestTimesQueue.get(RS);
        queue.add(t);
//        requestTimesQueue.put(RS,queue);
        queue.remove();
        for (Map.Entry<String, Queue<Integer>> entry : requestTimesQueue.entrySet()) {
            System.out.println("RS: " + RS + " and the time queue is: " +  requestTimesQueue.get(RS));
        }
//            requestTimesQueue.put(RS,queue);
    }
    public void flashQueue(Map<String ,Integer> map){
        synchronized (requestTimesQueue) {
            WriteToLog.write("requestTimesQueue is flashed!\n");
            for (Map.Entry<String, Integer> entry : map.entrySet()) {
                //以下六行是从Host转化为RegionServerName
                String RS = "";
                for (String name : MemoryInfo.getRegionServer().rsNameInCluster) {
                    System.out.print("");
                    if (name.contains(entry.getKey())) {
                        RS = name;
                        //System.out.print("RS:"+RS);
                    }
                }
                flashOneQueue(RS, entry.getValue());
            }
        }
    }
    public Map<String,Queue<Integer>> getRequestTimesQueue(){
        Map<String, Queue<Integer>> map= new HashMap<String, Queue<Integer>>();
        synchronized (requestTimesQueue){
            for (Map.Entry<String, Queue<Integer>> entry : requestTimesQueue.entrySet()) {
                Queue<Integer> queue1=new LinkedList<Integer>();
                queue1.addAll(entry.getValue());
                map.put(entry.getKey(),queue1);
            }
        }

        return map;

    }


//    public Queue<Integer> getTheRequestQueue(String RS){
//        return requestTimesQueue.get(RS);
//    }

}
